#include "db/datasegmentmanager.hxx"
#include "db/datastore_impl.hxx"
#include "db/dbpartition.hxx"
#include "db/iodatasegment.hxx"
#include "util/fs.hxx"
#include "util/lockguard.hxx"
#include "util/common.hxx"
#include "util/logger.hxx"
#include <vector>
#include <stdlib.h>



DataSegmentManager::DataSegmentManager(DataStoreImpl& ds, DbPartition& partition)
: m_segments(NULL), 
  m_partitionPath(partition.getPartitionPath()), 
  m_ds(&ds), 
  m_partition(&partition),
  m_activeSeg(NULL),
  m_activeSegCompact(NULL) {

}

DataSegmentManager::~DataSegmentManager() {

    if(m_segments == NULL) return;

    for(std::map<uint16_t, DataSegment*>::iterator iter = m_segments->begin(); 
            iter != m_segments->end(); ++iter) {
        if(iter->second) delete iter->second;
    }

    RWLockGuard lock(m_rwlockMap, 'w');
    m_segments->clear();
    lock.reset();

    if(m_segments) {delete m_segments; m_segments = NULL;}

    pthread_rwlock_destroy(&m_rwlockMap);
}

DbStatus DataSegmentManager::init() {

    DbStatus ret;

    int rc = pthread_rwlock_init(&m_rwlockMap, NULL);
    if(rc != 0) return DbStatus("DataSegmentManager::initn::pthread_rwlock_init", rc);

    m_segments = new std::map<uint16_t, DataSegment*>();
	ret = loadSegments();
	
    return ret;
}

DbStatus DataSegmentManager::loadSegments() {

    DbStatus ret;

	std::vector<std::string> children;
	FsUtility::getChildren(m_partitionPath, children);
	for(int i = 0; i < children.size(); ++i) {
        
        if(isdigit(children[i][0]) == false) continue;
		std::string segmentPath = m_partitionPath + "/" + children[i];
	        	
		if(!FsUtility::isDir(segmentPath)) {

			//TODO, check whether the segment name is valid
			uint16_t segID = atoi(children[i].c_str());
			
			DataSegment* seg = new IODataSegment(segID, *m_ds, *m_partition, m_ds->getSegCapacity());
			ret = seg->init();
			if(!ret.isok()) {delete seg; return ret;}

			(*m_segments)[segID] = seg;
		}
	}


	if(!m_segments->empty()) m_activeSeg = m_segments->rbegin()->second;
	else {
	
		ret = nextSegment(m_activeSeg); if(!ret.isok()) return ret; 
	}

    m_activeSeg->setMode(DataSegment::READ_WRITE);
	
	return ret;

}


DbStatus DataSegmentManager::flush() {

    DbStatus ret;

    RWLockGuard lock(m_rwlockMap, 'r');

    //flush the header for segments which changedSize is not 0
    for(std::map<uint16_t, DataSegment*>::iterator iter = m_segments->begin();
            iter != m_segments->end(); ++iter) {

        ret = (*iter).second->flush();

        if(!ret.isok()) break;
    }
 
	return ret; 
}

DbStatus DataSegmentManager::get(uint16_t segID, 
                                 uint32_t offset, 
                                 uint32_t& dataSize, 
                                 std::string& result) {

	DataSegment* seg = getSegment(segID); 
	if(seg == NULL) {
        std::string msg = "DataSegmentManager::get segID: " + utility::itoa(segID);
        return DbStatus(msg, DbStatus::DB_SEGMENTNOTFOUND);
    }

    //read my conflict with sync which may remove the segment, so put a lock
    //here
    RWLockGuard lock(m_rwlockMap, 'r');

	return seg->read(offset, dataSize, result);
}


DbStatus DataSegmentManager::get(uint16_t segID, 
                                 uint32_t offset, 
                                 uint32_t& dataSize, 
                                 Dbt& dbt) {

    DataSegment* seg = getSegment(segID);
    if(seg == NULL) {
        std::string msg = "DataSegmentManager::get segID: " + utility::itoa(segID);
        assert(false);
        return DbStatus(msg, DbStatus::DB_SEGMENTNOTFOUND);
    }

    //read my conflict with sync which may remove the segment, so put a lock
    //here
    RWLockGuard lock(m_rwlockMap, 'r');

    return seg->read(offset, dataSize, dbt);
}

//loadsize includes the size of the length and the sie of data
DbStatus DataSegmentManager::put(std::string& data, 
                                 uint16_t& segID, 
                                 uint32_t& offset, 
                                 uint32_t& loadSize) {

    DbStatus ret;

	if(m_activeSegCompact) ret = m_activeSegCompact->append(data, offset);
	if(m_activeSegCompact == NULL || ret.getCode() == DbStatus::DB_SEGMENTOVERFLOW) {

		DataSegment* seg;
        ret = nextSegment(seg); if(!ret.isok()) return ret;

		if(m_activeSegCompact) {

            m_activeSegCompact->flush();
            m_activeSegCompact->setMode(DataSegment::READ_ONLY);
            m_activeSegCompact->closeWriter();
        }

		m_activeSegCompact = seg;
        m_activeSegCompact->setMode(DataSegment::READ_WRITE);

		ret = m_activeSegCompact->append(data, offset);
	}

    segID = m_activeSegCompact->getSegId();

    uint32_t payloadLen = sizeof(uint32_t) + data.size();
    loadSize = m_activeSegCompact->changeDataSize(payloadLen);

	return ret;
}


DbStatus DataSegmentManager::put(Dbt& dbt, uint16_t& segID, uint32_t& offset, uint32_t& loadSize) {

    DbStatus ret;

    ret = m_activeSeg->append(dbt, offset);
    if(ret.getCode() == DbStatus::DB_SEGMENTOVERFLOW) {

        DataSegment* seg;
        ret = nextSegment(seg); if(!ret.isok()) return ret;

        if(m_activeSeg) {
//static utility::PerformanceMeasure sflush("seg flush"); sflush.start();
            m_activeSeg->flush();
//sflush.end();
            m_activeSeg->setMode(DataSegment::READ_ONLY);
            m_activeSeg->closeWriter();
        }

        m_activeSeg = seg;
        m_activeSeg->setMode(DataSegment::READ_WRITE);

        ret = m_activeSeg->append(dbt, offset);

    }

    segID = m_activeSeg->getSegId();

    uint32_t payloadLen = sizeof(uint32_t) + dbt.getDatalen();
    loadSize = m_activeSeg->changeDataSize(payloadLen);

    return ret;

}

//get doesn't need a lock
DataSegment* DataSegmentManager::getSegment(uint16_t segID) {

	DataSegment* seg = NULL;

    RWLockGuard lock(m_rwlockMap, 'r');
    std::map<uint16_t, DataSegment*>::iterator iter = m_segments->find(segID);
	if(iter != m_segments->end()) seg = iter->second;

	return seg;
}


uint16_t DataSegmentManager::nextSegID() {

    uint16_t segID = 0;
    if(m_activeSeg) segID = m_activeSeg->getSegId();
    segID++;

    int i = 0;
    for(; i <= 0xFFFF && (segID == 0 || m_segments->count(segID) > 0); segID++, ++i);
        
    return segID;
}


//nextSegment is the only function updating m_segments, to avoid using a clock, 
//nextSegment does a copy and update the copy and then set m_segments to the
//copy
DbStatus DataSegmentManager::nextSegment(DataSegment*& seg) {

    DbStatus ret;

    RWLockGuard lock(m_rwlockMap, 'w');

	uint16_t segID = nextSegID();

	seg = new IODataSegment(segID, *m_ds, *m_partition, m_ds->getSegCapacity());
	seg->setMode(DataSegment::READ_WRITE);
    ret =  seg->init(); if(!ret.isok()) return ret;

    (*m_segments)[segID] = seg;

    return ret;
}


std::set<uint16_t> DataSegmentManager::getSegmentIDsForCompact(double validDataRatio) {

    RWLockGuard lock(m_rwlockMap, 'r');
    
    std::set<uint16_t> s;
    for(std::map<uint16_t, DataSegment*>::iterator iter = m_segments->begin();
            iter != m_segments->end(); ++iter) {
        
        //the active segment should be excluded 
        if(iter->second == m_activeSeg || iter->second == m_activeSegCompact || !iter->second->isAlive()) continue;

        double rate =
            (double)iter->second->getActualDataSize()/(double)iter->second->getCapacity();

        if(rate <= validDataRatio) { 
            s.insert(iter->second->getSegId());
            Logger::log("segment %d: %f", iter->second->getSegId(), rate);
        }
    }

    return s;
}


DbStatus DataSegmentManager::discardSegment(uint16_t segID) {

    DbStatus ret;

    DataSegment* seg = getSegment(segID);

    RWLockGuard lock(m_rwlockMap, 'w');
    m_segments->erase(segID);

    std::string fname = seg->getPath() + "/" + utility::itoa(segID);
    delete seg;

    //TODO: reuse the segments
    ret = FsUtility::deleteFile(fname);
    Logger::log("delete segment %s", fname.c_str());

    return ret;
}


std::set<uint16_t> DataSegmentManager::getSegmentIDsNotAlive() {

    RWLockGuard lock(m_rwlockMap, 'r');

    std::set<uint16_t> s;
    for(std::map<uint16_t, DataSegment*>::iterator iter = m_segments->begin();
            iter != m_segments->end(); ++iter) {

        //the active segment should be excluded
        if(iter->second == m_activeSeg) continue;

        if( iter->second->isAlive() == false ) 
            s.insert(iter->second->getSegId());
    }

    return s;
}

